package net.bwie.vehicle.job;


import com.alibaba.fastjson.JSON;
import net.bwie.realtime.jtp.common.utils.KafkaUtil;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;
/*
*
* 数据清洗
* */
public class carDataJob {
    public static void main(String[] args) throws Exception {


        // TODO: 2025/5/26 执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        // TODO: 2025/5/26 获取kafka数据
        DataStream<String> kafkaDataStream = KafkaUtil.consumerKafka(env, "vehicle-data");
        kafkaDataStream.print();
        // TODO: 2025/5/26 数据清洗
        DataStream<String> jsonStream =cleanData(kafkaDataStream);
        // TODO: 2025/5/27 输出主流
        KafkaUtil.producerKafka(jsonStream,"car-vehicle-data");
        // TODO: 2025/5/26 触发执行
        env.execute("carDataJob");

    }

    private static DataStream<String> cleanData(DataStream<String> kafkaDataStream) {
        final OutputTag<String> dirtyTag = new OutputTag<String>("Dirty-data"){};
        SingleOutputStreamOperator<String> process = kafkaDataStream.process(new ProcessFunction<String, String>() {
            @Override
            public void processElement(String value, Context ctx, Collector<String> out) throws Exception {
                // TODO: 2025/5/26 清洗数据
                try {
                    // TODO: 2025/5/26 解析json数据
                JSON.parseObject(value);
                out.collect(value);
                }catch (Exception e){
                    ctx.output(dirtyTag,value);
                }

            }
        });
        DataStream<String> sideOutput = process.getSideOutput(dirtyTag);
        KafkaUtil.producerKafka(sideOutput,"Dirty-data");
       return process;
    }

}
